{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%run startup.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "application/javascript": [
       "$.getScript('./assets/js/ipython_notebook_toc.js')"
      ],
      "text/plain": [
       "<IPython.core.display.Javascript object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%javascript\n",
    "$.getScript('./assets/js/ipython_notebook_toc.js')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A Decision Tree of Observable Operators\n",
    "\n",
    "## Part 1: NEW Observables.\n",
    "\n",
    "> source: http://reactivex.io/documentation/operators.html#tree.  \n",
    "> (transcribed to RxPY 1.5.7, Py2.7 / 2016-12, Gunther Klessinger, [axiros](http://www.axiros.com))  \n",
    "\n",
    "**This tree can help you find the ReactiveX Observable operator you’re looking for.**  \n",
    "\n",
    "<h2 id=\"tocheading\">Table of Contents</h2>\n",
    "<div id=\"toc\"></div>\n",
    "\n",
    "## Usage\n",
    "\n",
    "There are no configured behind the scenes imports or code except [`startup.py`](./edit/startup.py), which defines output helper functions, mainly:\n",
    "\n",
    "- `rst, reset_start_time`: resets a global timer, in order to have use cases starting from 0.\n",
    "- `subs(observable)`: subscribes to an observable, printing notifications with time, thread, value\n",
    "\n",
    "\n",
    "All other code is explicitly given in the notebook.  \n",
    "Since all initialisiation of tools is in the first cell, you always have to run the first cell after ipython kernel restarts.  \n",
    "**All other cells are autonmous.**\n",
    "\n",
    "In the use case functions, in contrast to the official examples we simply use **`rand`** quite often (mapped to `randint(0, 100)`), to demonstrate when/how often observable sequences are generated and when their result is buffered for various subscribers.  \n",
    "*When in doubt then run the cell again, you might have been \"lucky\" and got the same random.*\n",
    "\n",
    "### RxJS\n",
    "The (bold printed) operator functions are linked to the [official documentation](http://reactivex.io/documentation/operators.html#tree) and created roughly analogous to the **RxJS** examples. The rest of the TOC lines links to anchors within the notebooks. \n",
    "\n",
    "### Output\n",
    "When the output is not in marble format we display it like so:\n",
    "\n",
    "```\n",
    "new subscription on stream 276507289 \n",
    "\n",
    "   3.4  M [next]    1.4: {'answer': 42}\n",
    "   3.5 T1 [cmpl]    1.6: fin\n",
    "   \n",
    "```\n",
    "where the lines are syncronously `print`ed as they happen.  \"M\" and \"T1\" would be thread names (\"M\" is main thread).  \n",
    "For each use case in `reset_start_time()` (alias `rst`), a global timer is set to 0 and we show the offset to it, in *milliseconds* & with one decimal value and also the offset to the start of stream subscription. In the example 3.4, 3.5 are millis since global counter reset, while 1.4, 1.6 are offsets to start of subscription.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# I want to create a **NEW** Observable..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ... that emits a particular item: **[just](http://reactivex.io/documentation/operators/just.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== return_value ==========\n",
      "\n",
      "module rx.linq.observable.returnvalue\n",
      "@extensionclassmethod(Observable, alias=\"just\")\n",
      "def return_value(cls, value, scheduler=None):\n",
      "    Returns an observable sequence that contains a single element,\n",
      "    using the specified scheduler to send out observer messages.\n",
      "    There is an alias called 'just'.\n",
      "\n",
      "    example\n",
      "    res = rx.Observable.return(42)\n",
      "    res = rx.Observable.return(42, rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    value -- Single element in the resulting observable sequence.\n",
      "    scheduler -- [Optional] Scheduler to send the single element on. If\n",
      "        not specified, defaults to Scheduler.immediate.\n",
      "\n",
      "    Returns an observable sequence containing the single specified\n",
      "    element.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.8     M New subscription on stream 273460685\n",
      "   2.8     M [next]    0.9: {'answer': 66}\n",
      "   3.3     M [cmpl]    1.5: fin\n",
      "\n",
      " 504.5     M New subscription on stream 273460685\n",
      " 505.0     M [next]    0.3: {'answer': 66}\n",
      " 505.1     M [cmpl]    0.4: fin\n",
      "\n",
      " 505.5     M New subscription on stream 272024237\n",
      " 506.3     M [next]    0.7: 132\n",
      " 506.8     M [cmpl]    1.1: fin\n"
     ]
    }
   ],
   "source": [
    "reset_start_time(O.just)\n",
    "stream = O.just({'answer': rand()})\n",
    "disposable = subs(stream)\n",
    "sleep(0.5)\n",
    "disposable = subs(stream) # same answer\n",
    "# all stream ops work, its a real stream:\n",
    "disposable = subs(stream.map(lambda x: x.get('answer', 0) * 2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ..that was returned from a function *called at subscribe-time*: **[start](http://reactivex.io/documentation/operators/start.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There is a little API difference to RxJS, see Remarks:\n",
      "\n",
      "\n",
      "\n",
      "========== start ==========\n",
      "\n",
      "module rx.linq.observable.start\n",
      "@extensionclassmethod(Observable)\n",
      "def start(cls, func, scheduler=None):\n",
      "    Invokes the specified function asynchronously on the specified\n",
      "    scheduler, surfacing the result through an observable sequence.\n",
      "\n",
      "    Example:\n",
      "    res = rx.Observable.start(lambda: pprint('hello'))\n",
      "    res = rx.Observable.start(lambda: pprint('hello'), rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    func -- {Function} Function to run asynchronously.\n",
      "    scheduler -- {Scheduler} [Optional] Scheduler to run the function on. If\n",
      "        not specified, defaults to Scheduler.timeout.\n",
      "\n",
      "    Returns {Observable} An observable sequence exposing the function's\n",
      "    result value, or an exception.\n",
      "\n",
      "    Remarks:\n",
      "    The function is called immediately, not during the subscription of the\n",
      "    resulting sequence. Multiple subscriptions to the resulting sequence can\n",
      "    observe the function's result.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.7    T4 function called   3.2     M New subscription on stream 274466149\n",
      "\n",
      "   3.7     M [next]    0.4: 43\n",
      "   3.8     M [cmpl]    0.5: fin\n",
      "\n",
      "   4.7     M New subscription on stream 274466149\n",
      "   5.1     M [next]    0.2: 43\n",
      "   5.3     M [cmpl]    0.4: fin\n",
      "\n",
      "\n",
      "========== Exceptions are handled correctly (an observable should never except): ==========\n",
      "\n",
      "\n",
      "   6.9     M New subscription on stream 274466197\n",
      "   7.5     M [err ]    0.5: integer division or modulo by zero\n",
      "\n",
      "   8.4     M New subscription on stream 274466197\n",
      "   8.9     M [err ]    0.3: integer division or modulo by zero\n"
     ]
    }
   ],
   "source": [
    "print('There is a little API difference to RxJS, see Remarks:\\n')\n",
    "rst(O.start)\n",
    "\n",
    "def f():\n",
    "    log('function called')\n",
    "    return rand()\n",
    "\n",
    "stream = O.start(func=f)\n",
    "d = subs(stream)\n",
    "d = subs(stream)\n",
    "\n",
    "header(\"Exceptions are handled correctly (an observable should never except):\")\n",
    "\n",
    "def breaking_f():    \n",
    "    return 1 / 0\n",
    "\n",
    "stream = O.start(func=breaking_f)\n",
    "d = subs(stream)\n",
    "d = subs(stream)\n",
    "\n",
    "\n",
    "\n",
    "# startasync: only in python3 and possibly here(?) http://www.tornadoweb.org/en/stable/concurrent.html#tornado.concurrent.Future\n",
    "#stream = O.start_async(f)\n",
    "#d = subs(stream)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ..that was returned from an Action, Callable, Runnable, or something of that sort, called at subscribe-time: **[from](http://reactivex.io/documentation/operators/from.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== from_iterable ==========\n",
      "\n",
      "module rx.linq.observable.fromiterable\n",
      "@extensionclassmethod(Observable, alias=[\"from_\", \"from_list\"])\n",
      "def from_iterable(cls, iterable, scheduler=None):\n",
      "    Converts an array to an observable sequence, using an optional\n",
      "    scheduler to enumerate the array.\n",
      "\n",
      "    1 - res = rx.Observable.from_iterable([1,2,3])\n",
      "    2 - res = rx.Observable.from_iterable([1,2,3], rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    :param Observable cls: Observable class\n",
      "    :param Scheduler scheduler: [Optional] Scheduler to run the\n",
      "        enumeration of the input sequence on.\n",
      "\n",
      "    :returns: The observable sequence whose elements are pulled from the\n",
      "        given enumerable sequence.\n",
      "    :rtype: Observable\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.3     M New subscription on stream 274466081\n",
      "   3.8     M [next]    0.4: 1\n",
      "   4.1     M [next]    0.7: 2\n",
      "   4.6     M [next]    1.1: 95\n",
      "   4.8     M [cmpl]    1.4: fin\n",
      "\n",
      "   5.4     M New subscription on stream 274466125\n",
      "   5.6     M [next]    0.2: 29\n",
      "   6.0     M [next]    0.6: 29\n",
      "   6.2     M [next]    0.8: 15\n",
      "   6.3     M [cmpl]    0.9: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.from_iterable)\n",
    "def f():\n",
    "    log('function called')\n",
    "    return rand()\n",
    "# aliases: O.from_, O.from_list\n",
    "# 1.: From a tuple:\n",
    "stream = O.from_iterable((1,2,rand()))\n",
    "d = subs(stream)\n",
    "# d = subs(stream) # same result\n",
    "\n",
    "# 2. from a generator\n",
    "gen = (rand() for j in range(3))\n",
    "stream = O.from_iterable(gen)\n",
    "d = subs(stream)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== from_callback ==========\n",
      "\n",
      "module rx.linq.observable.fromcallback\n",
      "@extensionclassmethod(Observable)\n",
      "def from_callback(cls, func, selector=None):\n",
      "    Converts a callback function to an observable sequence.\n",
      "\n",
      "    Keyword arguments:\n",
      "    func -- {Function} Function with a callback as the last parameter to\n",
      "        convert to an Observable sequence.\n",
      "    selector -- {Function} [Optional] A selector which takes the arguments\n",
      "        from the callback to produce a single item to yield on next.\n",
      "\n",
      "    Returns {Function} A function, when executed with the required\n",
      "    parameters minus the callback, produces an Observable sequence with a\n",
      "    single value of the arguments to the callback as a list.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   4.6     M New subscription on stream 272024249\n",
      "   5.9     M called f\n"
     ]
    }
   ],
   "source": [
    "rst(O.from_callback)\n",
    "# in my words: In the on_next of the subscriber you'll have the original arguments,\n",
    "# potentially objects, e.g. user original http requests.\n",
    "# i.e. you could merge those with the result stream of a backend call to\n",
    "# a webservice or db and send the request.response back to the user then.\n",
    "\n",
    "def g(f, a, b):\n",
    "    f(a, b)\n",
    "    log('called f')\n",
    "stream = O.from_callback(lambda a, b, f: g(f, a, b))('fu', 'bar')\n",
    "d = subs(stream.delay(200))\n",
    "# d = subs(stream.delay(200)) # does NOT work\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...after a specified delay: **[timer](http://reactivex.io/documentation/operators/timer.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "   0.8     M New subscription on stream 274470005\n",
      "\n",
      "   3.4     M New subscription on stream 274470005\n"
     ]
    }
   ],
   "source": [
    "rst()\n",
    "# start a stream of 0, 1, 2, .. after 200 ms, with a delay of 100 ms:\n",
    "stream = O.timer(200, 100).time_interval()\\\n",
    "    .map(lambda x: 'val:%s dt:%s' % (x.value, x.interval))\\\n",
    "    .take(3)\n",
    "d = subs(stream, name='observer1')\n",
    "# intermix directly with another one\n",
    "d = subs(stream, name='observer2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...that emits a sequence of items repeatedly: **[repeat](http://reactivex.io/documentation/operators/repeat.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== repeat ==========\n",
      "\n",
      "module rx.linq.observable.repeat\n",
      "@extensionclassmethod(Observable)\n",
      "def repeat(cls, value=None, repeat_count=None, scheduler=None):\n",
      "    Generates an observable sequence that repeats the given element the\n",
      "    specified number of times, using the specified scheduler to send out\n",
      "    observer messages.\n",
      "\n",
      "    1 - res = rx.Observable.repeat(42)\n",
      "    2 - res = rx.Observable.repeat(42, 4)\n",
      "    3 - res = rx.Observable.repeat(42, 4, Rx.Scheduler.timeout)\n",
      "    4 - res = rx.Observable.repeat(42, None, Rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    value -- Element to repeat.\n",
      "    repeat_count -- [Optional] Number of times to repeat the element. If not\n",
      "        specified, repeats indefinitely.\n",
      "    scheduler -- Scheduler to run the producer loop on. If not specified,\n",
      "        defaults to ImmediateScheduler.\n",
      "\n",
      "    Returns an observable sequence that repeats the given element the\n",
      "    specified number of times.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.0     M New subscription on stream 274473961\n",
      "   2.9     M [next]    0.9: {'rand': 1482335562.344726}\n",
      "   4.5     M [next]    2.4: {'rand': 1482335562.344726}\n",
      "   5.1     M [next]    3.0: {'rand': 1482335562.344726}\n",
      "   5.2     M [cmpl]    3.1: fin\n",
      "\n",
      "\n",
      "========== do while: ==========\n",
      "\n",
      "\n",
      "   6.8     M New subscription on stream 273460681\n",
      "   7.5     M [next]    0.5: 42\n",
      "   8.7     M [next]    1.7: 42\n",
      "   9.2     M [cmpl]    2.2: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.repeat)\n",
    "# repeat is over *values*, not function calls. Use generate or create for function calls!\n",
    "subs(O.repeat({'rand': time.time()}, 3))\n",
    "\n",
    "header('do while:')\n",
    "l = []\n",
    "def condition(x):\n",
    "    l.append(1)\n",
    "    return True if len(l) < 2 else False\n",
    "stream = O.just(42).do_while(condition)\n",
    "d = subs(stream)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...from scratch, with custom logic and cleanup (calling a function again and again): **[create](http://reactivex.io/documentation/operators/create.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== create ==========\n",
      "\n",
      "module rx.linq.observable.create\n",
      "@extensionclassmethod(Observable, alias=\"create_with_disposable\")\n",
      "def create(cls, subscribe):\n",
      "    n.a.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.4     M New subscription on stream 273454757\n",
      "   3.9     M cleaning up...\n",
      "\n",
      "   4.5     M New subscription on stream 273454757\n",
      "   5.8     M cleaning up...\n",
      " 131.3    T6 [next]  202.3: ['fu', 'bar']\n",
      " 131.7    T6 [cmpl]  202.7: fin\n",
      " 142.0    T7 [next]  202.4: val:0 dt:0:00:00.202066 (observer1)\n",
      " 144.0    T8 [next]  201.8: val:0 dt:0:00:00.201505 (observer2)\n",
      " 208.2    T9 [next]  205.7: 59\n",
      " 208.8    T9 [next]  206.3: 68\n",
      " 209.2    T9 [cmpl]  206.7: fin\n",
      " 209.6   T10 [next]  204.9: 84\n",
      " 210.0   T10 [next]  205.3: 79\n",
      " 210.2   T10 [cmpl]  205.4: fin\n",
      " 246.3   T12 [next]  304.1: val:1 dt:0:00:00.102253 (observer2)\n",
      " 247.0   T11 [next]  307.4: val:1 dt:0:00:00.104979 (observer1)\n",
      " 345.7   T14 [next]  406.1: val:2 dt:0:00:00.098724 (observer1)\n",
      " 346.0   T14 [cmpl]  406.4: fin (observer1)\n",
      " 348.3   T13 [next]  406.2: val:2 dt:0:00:00.102073 (observer2)\n",
      " 348.5   T13 [cmpl]  406.3: fin (observer2)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "MainThread:*** Exception: float division by zero\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "   0.6     M New subscription on stream 274475781\n",
      "   1.2     M [next]    0.3: 0 0.5 (observer hash: 272024249)\n",
      "   1.6     M [next]    0.7: 1 1.0 (observer hash: 272024249)\n",
      "   2.2     M [err ]    1.3: float division by zero\n",
      "\n",
      "   2.9     M New subscription on stream 274475781\n",
      "   3.2     M [next]    0.2: 0 -1.0 (observer hash: 272024253)\n",
      "   3.5     M [next]    0.5: 1 -0.5 (observer hash: 272024253)\n",
      "   4.0     M [next]    1.0: 2 -0.333333333333 (observer hash: 272024253)\n",
      "   4.1     M [cmpl]    1.0: fin\n",
      "(where are the first two values?)\n",
      "\n",
      "   0.7     M New subscription on stream 273460701\n",
      "   3.8     M [err ]    2.9: float division by zero\n",
      "\n",
      "   4.4     M New subscription on stream 273460701\n"
     ]
    }
   ],
   "source": [
    "rx = O.create\n",
    "rst(rx)\n",
    "\n",
    "def f(obs):\n",
    "    # this function is called for every observer\n",
    "    obs.on_next(rand())\n",
    "    obs.on_next(rand())\n",
    "    obs.on_completed()\n",
    "    def cleanup():\n",
    "        log('cleaning up...')\n",
    "    return cleanup\n",
    "stream = O.create(f).delay(200) # the delay causes the cleanup called before the subs gets the vals\n",
    "d = subs(stream)\n",
    "d = subs(stream)\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "sleep(0.5)\n",
    "rst(title='Exceptions are handled nicely')\n",
    "l = []\n",
    "def excepting_f(obs):\n",
    "    for i in range(3):\n",
    "        l.append(1)\n",
    "        obs.on_next('%s %s (observer hash: %s)' % (i, 1. / (3 - len(l)), hash(obs) ))\n",
    "    obs.on_completed()\n",
    "\n",
    "stream = O.create(excepting_f)\n",
    "d = subs(stream)\n",
    "d = subs(stream)\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "rst(title='Feature or Bug?')\n",
    "print('(where are the first two values?)')\n",
    "l = []\n",
    "def excepting_f(obs):\n",
    "    for i in range(3):\n",
    "        l.append(1)\n",
    "        obs.on_next('%s %s (observer hash: %s)' % (i, 1. / (3 - len(l)), hash(obs) ))\n",
    "    obs.on_completed()\n",
    "\n",
    "stream = O.create(excepting_f).delay(100)\n",
    "d = subs(stream)\n",
    "d = subs(stream)\n",
    "# I think its an (amazing) feature, preventing to process functions results of later(!) failing functions\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== generate ==========\n",
      "\n",
      "module rx.linq.observable.generate\n",
      "@extensionclassmethod(Observable)\n",
      "def generate(cls, initial_state, condition, iterate, result_selector, scheduler=None):\n",
      "    Generates an observable sequence by running a state-driven loop\n",
      "    producing the sequence's elements, using the specified scheduler to\n",
      "    send out observer messages.\n",
      "\n",
      "    1 - res = rx.Observable.generate(0,\n",
      "        lambda x: x < 10,\n",
      "        lambda x: x + 1,\n",
      "        lambda x: x)\n",
      "    2 - res = rx.Observable.generate(0,\n",
      "        lambda x: x < 10,\n",
      "        lambda x: x + 1,\n",
      "        lambda x: x,\n",
      "        Rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    initial_state -- Initial state.\n",
      "    condition -- Condition to terminate generation (upon returning False).\n",
      "    iterate -- Iteration step function.\n",
      "    result_selector -- Selector function for results produced in the\n",
      "        sequence.\n",
      "    scheduler -- [Optional] Scheduler on which to run the generator loop.\n",
      "        If not provided, defaults to CurrentThreadScheduler.\n",
      "\n",
      "    Returns the generated sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   4.8     M New subscription on stream 274475993\n",
      "   5.7     M [next]    0.7: 0\n",
      "   6.4     M [next]    1.4: 2.2\n",
      "   6.6     M [next]    1.6: 4.4\n",
      "   7.1     M [next]    2.1: 6.6\n",
      "   7.3     M [cmpl]    2.3: fin\n"
     ]
    }
   ],
   "source": [
    "rx = O.generate\n",
    "rst(rx)\n",
    "\"\"\"The basic form of generate takes four parameters:\n",
    "\n",
    "the first item to emit\n",
    "a function to test an item to determine whether to emit it (true) or terminate the Observable (false)\n",
    "a function to generate the next item to test and emit based on the value of the previous item\n",
    "a function to transform items before emitting them\n",
    "\"\"\"\n",
    "def generator_based_on_previous(x): return x + 1.1\n",
    "def doubler(x): return 2 * x\n",
    "d = subs(rx(0, lambda x: x < 4, generator_based_on_previous, doubler))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== generate_with_relative_time ==========\n",
      "\n",
      "module rx.linq.observable.generatewithrelativetime\n",
      "@extensionclassmethod(Observable)\n",
      "def generate_with_relative_time(cls, initial_state, condition, iterate,\n",
      "    Generates an observable sequence by iterating a state from an\n",
      "    initial state until the condition fails.\n",
      "\n",
      "    res = source.generate_with_relative_time(0,\n",
      "        lambda x: True,\n",
      "        lambda x: x + 1,\n",
      "        lambda x: x,\n",
      "        lambda x: 500)\n",
      "\n",
      "    initial_state -- Initial state.\n",
      "    condition -- Condition to terminate generation (upon returning false).\n",
      "    iterate -- Iteration step function.\n",
      "    result_selector -- Selector function for results produced in the\n",
      "        sequence.\n",
      "    time_selector -- Time selector function to control the speed of values\n",
      "        being produced each iteration, returning integer values denoting\n",
      "        milliseconds.\n",
      "    scheduler -- [Optional] Scheduler on which to run the generator loop.\n",
      "        If not specified, the timeout scheduler is used.\n",
      "\n",
      "    Returns the generated sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   4.7     M New subscription on stream 274475933\n"
     ]
    }
   ],
   "source": [
    "rx = O.generate_with_relative_time\n",
    "rst(rx)\n",
    "stream = rx(1, lambda x: x < 4, lambda x: x + 1, lambda x: x, lambda t: 100)\n",
    "d = subs(stream)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...for each observer that subscribes OR according to a condition at subscription time: **[defer / if_then](http://reactivex.io/documentation/operators/defer.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== defer ==========\n",
      "\n",
      "module rx.linq.observable.defer\n",
      "@extensionclassmethod(Observable)\n",
      "def defer(cls, observable_factory):\n",
      "    Returns an observable sequence that invokes the specified factory\n",
      "    function whenever a new observer subscribes.\n",
      "\n",
      "    Example:\n",
      "    1 - res = rx.Observable.defer(lambda: rx.Observable.from_([1,2,3]))\n",
      "\n",
      "    Keyword arguments:\n",
      "    :param types.FunctionType observable_factory: Observable factory function\n",
      "        to invoke for each observer that subscribes to the resulting sequence.\n",
      "\n",
      "    :returns: An observable sequence whose observers trigger an invocation\n",
      "    of the given observable factory function.\n",
      "    :rtype: Observable\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.7     M New subscription on stream 274475969\n",
      "   3.4     M [next]    0.6: 38\n",
      "   3.5     M [cmpl]    0.7: fin\n",
      "\n",
      "   4.4     M New subscription on stream 274475969\n",
      "   4.9     M [next]    0.4: 77\n",
      "   5.2     M [cmpl]    0.7: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.defer)\n",
    "# plural! (unique per subscription)\n",
    "streams = O.defer(lambda: O.just(rand()))\n",
    "d = subs(streams)\n",
    "d = subs(streams) # gets other values - created by subscription!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== if_then ==========\n",
      "\n",
      "module rx.linq.observable.ifthen\n",
      "@extensionclassmethod(Observable)\n",
      "def if_then(cls, condition, then_source, else_source=None, scheduler=None):\n",
      "    Determines whether an observable collection contains values.\n",
      "\n",
      "    Example:\n",
      "    1 - res = rx.Observable.if(condition, obs1)\n",
      "    2 - res = rx.Observable.if(condition, obs1, obs2)\n",
      "    3 - res = rx.Observable.if(condition, obs1, scheduler=scheduler)\n",
      "\n",
      "    Keyword parameters:\n",
      "    condition -- {Function} The condition which determines if the\n",
      "        then_source or else_source will be run.\n",
      "    then_source -- {Observable} The observable sequence or Promise that\n",
      "        will be run if the condition function returns true.\n",
      "    else_source -- {Observable} [Optional] The observable sequence or\n",
      "        Promise that will be run if the condition function returns False.\n",
      "        If this is not provided, it defaults to rx.Observable.empty\n",
      "    scheduler -- [Optional] Scheduler to use.\n",
      "\n",
      "    Returns an observable {Observable} sequence which is either the\n",
      "    then_source or else_source.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.3     M New subscription on stream 274480673\n",
      "   3.6     M [next]    0.2: 43\n",
      "   3.8     M [cmpl]    0.5: fin\n",
      "   4.0     M condition will now evaluate falsy:\n",
      "\n",
      "   4.4     M New subscription on stream 274480817\n",
      "   4.6     M [next]    0.2: 52\n",
      "   4.7     M [cmpl]    0.3: fin\n",
      "\n",
      "   5.2     M New subscription on stream 274480817\n",
      "   5.6     M [next]    0.2: 52\n",
      "   5.8     M [cmpl]    0.4: fin\n"
     ]
    }
   ],
   "source": [
    "# evaluating a condition at subscription time in order to decide which of two streams to take.\n",
    "rst(O.if_then)\n",
    "cond = True\n",
    "def should_run():\n",
    "    return cond\n",
    "streams = O.if_then(should_run, O.return_value(43), O.return_value(56))\n",
    "d = subs(streams)\n",
    "\n",
    "log('condition will now evaluate falsy:')\n",
    "cond = False\n",
    "streams = O.if_then(should_run, O.return_value(43), O.return_value(rand()))\n",
    "d = subs(streams)\n",
    "d = subs(streams)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...that emits a sequence of integers: **[range](http://reactivex.io/documentation/operators/range.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== range ==========\n",
      "\n",
      "module rx.linq.observable.range\n",
      "@extensionclassmethod(Observable)\n",
      "def range(cls, start, count, scheduler=None):\n",
      "    Generates an observable sequence of integral numbers within a\n",
      "    specified range, using the specified scheduler to send out observer\n",
      "    messages.\n",
      "\n",
      "    1 - res = Rx.Observable.range(0, 10)\n",
      "    2 - res = Rx.Observable.range(0, 10, rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    start -- The value of the first integer in the sequence.\n",
      "    count -- The number of sequential integers to generate.\n",
      "    scheduler -- [Optional] Scheduler to run the generator loop on. If not\n",
      "        specified, defaults to Scheduler.current_thread.\n",
      "\n",
      "    Returns an observable sequence that contains a range of sequential\n",
      "    integral numbers.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.9     M New subscription on stream 274475905\n",
      "   3.7     M [next]    0.4: 0\n",
      "   4.3     M [next]    1.0: 1\n",
      "   4.6     M [next]    1.3: 2\n",
      "   4.9     M [cmpl]    1.6: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.range)\n",
    "d = subs(O.range(0, 3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ...at particular intervals of time: **[interval](http://reactivex.io/documentation/operators/interval.html) **\n",
    "\n",
    "(you can `.publish()` it to get an easy \"hot\" observable)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== interval ==========\n",
      "\n",
      "module rx.linq.observable.interval\n",
      "@extensionclassmethod(Observable)\n",
      "def interval(cls, period, scheduler=None):\n",
      "    Returns an observable sequence that produces a value after each\n",
      "    period.\n",
      "\n",
      "    Example:\n",
      "    1 - res = rx.Observable.interval(1000)\n",
      "    2 - res = rx.Observable.interval(1000, rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    period -- Period for producing the values in the resulting sequence\n",
      "        (specified as an integer denoting milliseconds).\n",
      "    scheduler -- [Optional] Scheduler to run the timer on. If not specified,\n",
      "        rx.Scheduler.timeout is used.\n",
      "\n",
      "    Returns an observable sequence that produces a value after each period.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.2     M New subscription (14365) on stream 276610125\n",
      " 102.3    T8 [next]  100.9: 0:00:00.100623 0 -> 14365\n",
      " 208.2    T9 [next]  206.9: 0:00:00.105960 1 -> 14365\n",
      " 310.8   T10 [next]  309.5: 0:00:00.102625 2 -> 14365\n",
      " 311.1   T10 [cmpl]  309.8: fin -> 14365\n"
     ]
    }
   ],
   "source": [
    "rst(O.interval)\n",
    "d = subs(O.interval(100).time_interval()\\\n",
    "         .map(lambda x, v: '%(interval)s %(value)s' \\\n",
    "              % ItemGetter(x)).take(3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ...after a specified delay (see timer)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...that completes without emitting items: **[empty](http://reactivex.io/documentation/operators/empty-never-throw.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== empty ==========\n",
      "\n",
      "module rx.linq.observable.empty\n",
      "@extensionclassmethod(Observable)\n",
      "def empty(cls, scheduler=None):\n",
      "    Returns an empty observable sequence, using the specified scheduler\n",
      "    to send out the single OnCompleted message.\n",
      "\n",
      "    1 - res = rx.Observable.empty()\n",
      "    2 - res = rx.Observable.empty(rx.Scheduler.timeout)\n",
      "\n",
      "    scheduler -- Scheduler to send the termination call on.\n",
      "\n",
      "    Returns an observable sequence with no elements.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.9     M New subscription on stream 273460593\n",
      "   3.2     M [cmpl]    0.2: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.empty)\n",
    "d = subs(O.empty())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...that does nothing at all: **[never](http://reactivex.io/documentation/operators/empty-never-throw.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== never ==========\n",
      "\n",
      "   0.7   T18 [next]  104.4: 0 -1.0 (observer hash: 274473797)\n",
      "   1.1   T18 [next]  104.8: 1 -0.5 (observer hash: 274473797)module rx.linq.observable.never\n",
      "@extensionclassmethod(Observable)\n",
      "def never(cls):\n",
      "    Returns a non-terminating observable sequence, which can be used to\n",
      "    denote an infinite duration (e.g. when using reactive joins).\n",
      "\n",
      "    Returns an observable sequence whose observers will never get called.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.0   T18 [next]  105.7: 2 -0.333333333333 (observer hash: 274473797)\n",
      "\n",
      "   2.1   T18 [cmpl]  105.9: fin   2.7     M New subscription on stream 274473849\n",
      "\n"
     ]
    }
   ],
   "source": [
    "rst(O.never)\n",
    "d = subs(O.never())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...that excepts: **[throw](http://reactivex.io/documentation/operators/empty-never-throw.html) **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== throw ==========\n",
      "\n",
      "module rx.linq.observable.throw\n",
      "@extensionclassmethod(Observable, alias=\"throw_exception\")\n",
      "def throw(cls, exception, scheduler=None):\n",
      "    Returns an observable sequence that terminates with an exception,\n",
      "    using the specified scheduler to send out the single OnError message.\n",
      "\n",
      "    1 - res = rx.Observable.throw_exception(Exception('Error'))\n",
      "    2 - res = rx.Observable.throw_exception(Exception('Error'),\n",
      "                                            rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    exception -- An object used for the sequence's termination.\n",
      "    scheduler -- Scheduler to send the exceptional termination call on. If\n",
      "        not specified, defaults to ImmediateScheduler.\n",
      "\n",
      "    Returns the observable sequence that terminates exceptionally with the\n",
      "    specified exception object.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.8     M New subscription (23467) on stream 276521733\n",
      "   2.0     M [err ]    0.2: <type 'exceptions.ZeroDivisionError'> -> 23467\n"
     ]
    }
   ],
   "source": [
    "rst(O.throw)\n",
    "d = subs(O.throw(ZeroDivisionError))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
